package com.asap.demo.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
//import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.table.factories.TableSourceFactory;
/**
 * @author wangbh
 * @Description: test
 * @date 2021/7/9 15:16
 */
public class FlinkSQLDistinctExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment blinkStreamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
//		blinkStreamEnv.setParallelism(1);
        EnvironmentSettings blinkStreamSettings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        StreamTableEnvironment blinkStreamTableEnv = StreamTableEnvironment.create(blinkStreamEnv, blinkStreamSettings);

        String ddlSource = "CREATE TABLE asap_superset (\n" +
                "    SRC_PORT STRING,\n" +
                "    DST_PORT STRING,\n" +
                "    DST_IP STRING,\n" +
                "    SRC_IP STRING,\n" +
                "    CREATE_TIME TIMESTAMP(3),\n" +
                "    DEVICE_PARENT_TYPE STRING,\n" +
                "    SNOW_ID BIGINT, \n " +
                "    EVENT_THREE_TYPE_DESC STRING," +
                "    ts TIMESTAMP(3),\n" +
                "    pct AS PROCTIME(), \n" +
                "    WATERMARK FOR CREATE_TIME AS CREATE_TIME - INTERVAL '10' SECOND \n" +
                ") WITH (\n" +
                "    'connector.type' = 'kafka',\n" +
               "    'connector.version' = 'universal',\n" +
                "    'connector.topic' = 'flink_pressure_test',\n" +
                "    'connector.startup-mode' = 'latest-offset',\n" +
                "    'connector.properties.zookeeper.connect' = '192.168.1.25:1813',\n" +
                "    'connector.properties.bootstrap.servers' = '192.168.1.25:9093',\n" +
                "    'connector.properties.group.id' = 'distinct',\n" +
                "    'format.type' = 'json'\n" +
                ")";


        blinkStreamTableEnv.executeSql(ddlSource);
        String querySQL = "select * from asap_superset where DEVICE_PARENT_TYPE='WAF' AND EVENT_THREE_TYPE_DESC='端口扫描'";
        Table table = blinkStreamTableEnv.sqlQuery(querySQL);
        blinkStreamTableEnv.toRetractStream(table, Row.class).print("query==");

        String countSql = "select TUMBLE_START(CREATE_TIME, INTERVAL '5' MINUTE) AS cntStart,TUMBLE_END(CREATE_TIME, INTERVAL '5' MINUTE) AS cntEnd " +
                "from  asap_superset group by TUMBLE(CREATE_TIME, INTERVAL '5' MINUTE),SRC_PORT";
        Table countTable = blinkStreamTableEnv.sqlQuery(countSql);
        blinkStreamTableEnv.toRetractStream(countTable, Row.class).print("query0==");

//		String distinctSql = "select COUNT(DISTINCT(DST_IP)) " +
//				"from asap_superset " +
//				"where DEVICE_PARENT_TYPE='WAF' " +
//				"group by TUMBLE(CREATE_TIME, INTERVAL '15' MINUTE),SRC_PORT";
        String querySQL1 = "select collect(DISTINCT(DST_IP)) as DST_IP ,TUMBLE_START(CREATE_TIME, INTERVAL '5' MINUTE) AS cntStart," +
                "TUMBLE_END(CREATE_TIME, INTERVAL '5' MINUTE) AS cntEnd " +
                " from asap_superset " +
                "where (DEVICE_PARENT_TYPE='WAF' or DEVICE_PARENT_TYPE='FW') AND EVENT_THREE_TYPE_DESC='端口扫描' " +
                "group by TUMBLE(CREATE_TIME, INTERVAL '5' MINUTE),SRC_PORT ";
        Table table1 = blinkStreamTableEnv.sqlQuery(querySQL1);
        blinkStreamTableEnv.toRetractStream(table1, Row.class).print("query1==");

        String distinctSql = "select TUMBLE_START(CREATE_TIME, INTERVAL '5' MINUTE) AS cntStart," +
                "TUMBLE_END(CREATE_TIME, INTERVAL '5' MINUTE) AS cntEnd," +
                "collect(SNOW_ID) AS parent_uuids," +
                "collect(SRC_PORT) AS src_ports," +
                "collect(DST_IP) AS dst_ips," +
                "COUNT(DISTINCT(DST_IP)) distinct_count," +
                "COUNT(1), " +
                "MIN(pct) AS min_pct," +
                "MAX(pct) AS max_pct " +
                "from asap_superset " +
                "where (DEVICE_PARENT_TYPE='WAF' or DEVICE_PARENT_TYPE='FW') AND EVENT_THREE_TYPE_DESC='端口扫描' " +
                "group by TUMBLE(CREATE_TIME, INTERVAL '5' MINUTE),SRC_PORT " +
                "having COUNT(DISTINCT(DST_IP)) > 3";
        Table distinctTable = blinkStreamTableEnv.sqlQuery(distinctSql);
        blinkStreamTableEnv.toRetractStream(distinctTable, Row.class).print("Distinct alarm output==");

        blinkStreamEnv.execute("Blink Stream SQL distinct demo");

//		blinkStreamTableEnv.execute("Blink Stream SQL count/distinct demo");
    }
}
